package com.wj.opratorlog.cdc.debezium;/**
 * @author
 * @description
 * @create 2024-08-24-17:47
 **/

import lombok.extern.slf4j.Slf4j;

import java.util.Properties;

/**
 * @author cnry
 * @date 2024/8/24
 * @description
 *
 *
 *
 *   final Properties props = new Properties();
 *
 *         props.setProperty("name", "instala-core");
 *         props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
 *         //偏移量持久化,用来容错
 *         props.setProperty("offset.storage","org.apache.kafka.connect.storage.FileOffsetBackingStore");
 *         //偏移量持久化文件路径,默认/tmp/offsets.dat,如果路径配置不正确可能导致无法存储偏移量 可能会导致重复消费变更
 *         props.setProperty("offset.storage.file.filename", "D:/tmp/offsets.dat");
 *         //如果连接器重新启动，它将使用最后记录的偏移量来知道它应该恢复读取源信息中的哪个位置。
 *         props.setProperty("offset.flush.interval.ms", "60000");
 *
 *         //需要监听的数据库名称
 *         props.setProperty("database.whitelist", "report_sharing_center");
 *         //initial(默认) 	连接器执行数据库的初始一致性快照，快照完成后，连接器开始为后续数据库更改流式传输事件记录。
 *         //initial_only 	连接器只执行数据库的初始一致性快照，不允许捕获任何后续更改的事件。
 *         //schema_only 	连接器只捕获所有相关表的表结构，不捕获初始数据，但是会同步后续数据库的更改记录
 *         //schema_only_recovery 	设置此选项可恢复丢失或损坏的数据库历史主题(database.history.kafka.topic)。
 *         props.setProperty("snapshot.mode", "schema_only");
 *         //数据库地址
 *         props.setProperty("database.hostname", "xxx");
 *         //数据库端口
 *         props.setProperty("database.port","3306");
 *         //数据库用户名
 *         props.setProperty("database.user", "xxx");
 *         //数据库密码
 *         props.setProperty("database.password", "xxx");
 *         //server.id起到唯一标识作用,随意起
 *         props.setProperty("database.server.id", "xxx");
 *         //server.name起到唯一标识作用,随意起
 *         props.setProperty("database.server.name", "xxx");
 *
 *         //历史变更记录
 *         props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
 *         //历史变更记录存储位置
 *         props.setProperty("database.history.file.filename", "D:/tmp/dbhistory.dat");
 *
 *         //格式化日期
 *         props.setProperty("converters", "dateConverters");
 *         props.setProperty("dateConverters.type", "com.zy.data.sync.common.conver.MySqlDateTimeConverter");
 *         props.setProperty("dateConverters.format.date", "yyyy-MM-dd");
 *         props.setProperty("dateConverters.format.time", "HH:mm:ss");
 *         props.setProperty("dateConverters.format.datetime", "yyyy-MM-dd HH:mm:ss");
 *         props.setProperty("dateConverters.format.timestamp", "yyyy-MM-dd HH:mm:ss");
 *         props.setProperty("dateConverters.format.timestamp.zone", "UTC+8");
 *
 *         //全局读写锁,会影响在线业务,所以跳过锁设置
 *         props.setProperty("debezium.snapshot.locking.mode","none");
 *         //是否包含数据库表结构层面的变更,建议使用默认值true
 *         props.setProperty("include.schema.changes", "true");
 *
 *         //指定 BIGINT UNSIGNED 列应如何在更改事件中表示。可能的设置有
 *         //long使用 Java 的 表示值long，这可能无法提供精确度，但在消费者中易于使用。long通常是首选设置。
 *         props.setProperty("bigint.unsigned.handling.mode","long");
 *         //decimal类型转换为double
 *         props.setProperty("decimal.handling.mode","double");
 *
 *         DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
 *                 .using(props)
 *                 .notifying(record -> {
 *                     String recordStr = record.value();
 *                     if(StringUtils.isNotEmpty(recordStr)){
 *                         System.out.println("recordStr-->"+recordStr);
 *                         Map<String, Object> payload = DebeziumDataHande.getPayload(recordStr);
 *
 *                         //to do something
 *                         if(payload.containsKey("op")&&payload.get("op").equals("c")){
 *
 *                         }
 *
 *                     }
 *
 *                 }).build();
 *
 *         Executors.newSingleThreadExecutor().execute(engine);
 *
 *         // Run the engine asynchronously ...
 *         ExecutorService executor = Executors.newSingleThreadExecutor();
 *         executor.execute(engine);
 *
 * ————————————————
 *
 *                             版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
 *
 * 原文链接：https://blog.csdn.net/u013008898/article/details/132827141
 **/


@Slf4j
public class DebeziumCDC {

    public static Properties getProperties() {
        final Properties props = new Properties();
        /* begin connector properties */
        props.setProperty("database.hostname", "localhost"); // ip
        props.setProperty("database.port", "3306");               // 端口
        props.setProperty("database.user", "root");               // 用户名
        props.setProperty("database.password", "root");         // 密码
        props.setProperty("database.server.id", "1"); // value可以任意修改,别重复
        props.setProperty("database.server.name", "userservie");  // 可以任意修改,别重复
        props.setProperty("database.serverTimezone", "GMT"); // 时区
        // 1. 使用database.whitelist，只设置数据库（会通知全库的CDC信息）
        // 2. 使用table.whitelist，设置库名和表名（会通知单个库的单个表的CDC信息）
//        props.setProperty("database.whitelist", "userservie");    // 库名
//      props.setProperty("table.whitelist", "db_inventory_cdc.tb_products_cdc"); // 库.表名

        props.setProperty("name", "instala-core");
        props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", "D:\\cdcdata\\offsets.dat");
        props.setProperty("offset.flush.interval.ms", "60000");
        props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
        props.setProperty("database.history.file.filename", "D:\\cdcdata\\dbhistory.dat");

        props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.flush.interval.ms", String.valueOf(10000L));
        props.setProperty("key.converter.schemas.enable", "false");
        props.setProperty("value.converter.schemas.enable", "true");
        props.setProperty("include.schema.changes", "true");
        props.setProperty("tombstones.on.delete", "false");
//        props.setProperty("database.history", "io.debezium.relational.history.FileDatabaseHistory");
//        props.setProperty("database.history.store.only.monitored.tables.ddl", "true");
//        props.setProperty("database.history.instance.name", UUID.randomUUID().toString());

        return  props;

    }


}
